package com.wangke.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author wangke
 * @version v1.0
 * @项目名称: study
 * @描述:
 * @创建时间: 2019/4/22
 * @update [序号][日期YYYY-MM-DD] [更改人姓名][变更描述]
 *
 * 虽然设置了autocomit=false，依然会自动commit offset，这时候：
 * autocommit=false时，看你设置的ackMode是什么，如果是manual，那么就需要手动提交，是其他的值，onMessage处理完之后会自动提交。
 *actMode方式：
    RECORD
    每处理一条commit一次
    BATCH(默认)
    每次poll的时候批量提交一次，频率取决于每次poll的调用频率
    TIME
    每次间隔ackTime的时间去commit(跟auto commit interval有什么区别呢？)
    COUNT
    累积达到ackCount次的ack去commit
    COUNT_TIME
    ackTime或ackCount哪个条件先满足，就commit
    MANUAL
    listener负责ack，但是背后也是批量上去
    MANUAL_IMMEDIATE
    listner负责ack，每调用一次，就立即commit

 *
 *
 *
 *
 * 异常的话直接从onMessage里抛出去，这时候spring-kafka会看你的ackOnError设置的值，是true的话会提交，false的话不提交
 *
 */
@Component
@Slf4j
public class KafkaReceiver {

    @KafkaListener(topics = {"wkTest"},id = "0",groupId = "0")
    public void listen(ConsumerRecord<?, ?> record,Consumer consumer) {

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            consumer.commitSync(); //不提交 如果有其他消费者还会重复消费

            log.info("--------0--------- record =" + record);
            log.info("--------0---------- message =" + message);
        }

    }



    @KafkaListener(topics = {"wkTest"},id = "1",groupId = "0")
    public void listen1(ConsumerRecord<?, ?> record,Consumer consumer) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            consumer.commitSync(); //不提交 如果有其他消费者还会重复消费
            log.info("-------1---------- record =" + record);
            log.info("-------1----------- message =" + message);
        }

    }

}
